package com.sunwayworld.basemodule.common.asynctask.comsumer;

import com.sunwayworld.basemodule.common.asynctask.bean.LimsAsynctaskBean;
import com.sunwayworld.basemodule.common.asynctask.config.AsyncTaskExecutorConfig;
import com.sunwayworld.basemodule.common.asynctask.config.AsyncTaskStatusEnum;
import com.sunwayworld.basemodule.common.asynctask.config.LimsAsyncTaskExecutor;
import com.sunwayworld.basemodule.common.asynctask.service.LimsAsynctaskService;
import com.sunwayworld.basemodule.common.asynctask.util.AsyncTaskExecutorHelper;
import com.sunwayworld.framework.context.ApplicationContextHelper;
import com.sunwayworld.framework.database.sql.Order;
import com.sunwayworld.framework.mybatis.mapper.MapperParameter;
import com.sunwayworld.framework.mybatis.mapper.MatchPattern;
import com.sunwayworld.framework.mybatis.mapper.SearchFilter;
import com.sunwayworld.framework.utils.ObjectUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

@Component
public class AsyncTaskConsumer {
	private static final Logger logger = LogManager.getLogger(AsyncTaskConsumer.class);

	@Autowired
	private LimsAsynctaskService asynctaskService;
	@Autowired
	private AsyncTaskExecutorConfig taskExecutorConfig;
	@Autowired
	private AsyncTaskExecutorHelper asyncTaskExecutorHelper;

	/**
	 查异步任务，并加入线程池
	 */
	public void searchTaskAndExecute() {
		logger.info("searchTaskAndExecute");

		//主动去取任务
		//查找未设置serverId且状态是未分配、执行失败的，执行次数小于execMaxCount
		MapperParameter parameter = new MapperParameter();
		parameter.setFilter(SearchFilter.instance().match("execStatus", Arrays.asList(AsyncTaskStatusEnum.pendingAssign.status, AsyncTaskStatusEnum.executeFail.status)).filter(MatchPattern.OR));
		parameter.setOrderParam("createdTime","asc");
		parameter.put("ltExecMaxCount","1");
		List<LimsAsynctaskBean> existTasks = asynctaskService.selectList(parameter);
		if(ObjectUtils.isEmpty(existTasks)){
			return;
		}
		List<LimsAsynctaskBean> serverIdNullList = existTasks.stream().filter(g -> ObjectUtils.isEmpty(g.getServerId())).collect(Collectors.toList());
		//尝试把任务给自己
		executeTaskOnCurrentNode(serverIdNullList);
	}

	//将指定的任务加入线程池
	public void executeTaskOnCurrentNode(List<LimsAsynctaskBean> serverTasks) {
		if(taskExecutorConfig==null){
			return;
		}
		boolean enable = taskExecutorConfig.getEnable();
		String currentServerId = taskExecutorConfig.getCurrentServerId();
		if(!enable || ObjectUtils.isEmpty(currentServerId) || ObjectUtils.isEmpty(serverTasks)){
			return;
		}

		List<String> executorNames = serverTasks.stream().map(f -> f.getExecutorName()).distinct().collect(Collectors.toList());
		for (String executorName : executorNames) {
			if(ObjectUtils.isEmpty(executorName)){
				logger.error("executorName is null");
				continue;
			}

			List<LimsAsynctaskBean> executorTasks = serverTasks.stream().filter(f -> executorName.equals(f.getExecutorName())).collect(Collectors.toList());
			if(ObjectUtils.isEmpty(executorTasks)){
				continue;
			}

			boolean hasRegister = asyncTaskExecutorHelper.hasRegister(executorName);
			if(!hasRegister){
				logger.error("executorName={} is not hasRegister",executorName);
				continue;
			}

			LimsAsyncTaskExecutor threadPoolExecutor = asyncTaskExecutorHelper.getByExecutorName(executorName);
			if(threadPoolExecutor==null){
				logger.error("executor{} is not found",executorName);
				continue;
			}

			int waitQueueSize = threadPoolExecutor.getQueue().size();
			int activeCount = threadPoolExecutor.getActiveCount();
			int maximumPoolSize = threadPoolExecutor.getMaximumPoolSize();
			logger.info("{}:maximumPoolSize={},当前排队任务数:{},当前进行中任务数:{}",executorName,maximumPoolSize,waitQueueSize,activeCount);
			if(activeCount<maximumPoolSize){
				int addMax = maximumPoolSize-activeCount;
				//add to executor 只填满maxthread
				int add = 0;
				List<LimsAsynctaskBean> updateRunnings = new ArrayList<>();
				for (LimsAsynctaskBean serverTask : executorTasks) {
					add++;
					if(add<=addMax){
						updateRunnings.add(serverTask);
					}else{
						break;
					}
				}
				if(!ObjectUtils.isEmpty(updateRunnings)){
					PlatformTransactionManager transactionManager = ApplicationContextHelper.getBean(PlatformTransactionManager.class);
					//更新为执行中
					TransactionStatus transactionExecuting = transactionManager.getTransaction(new DefaultTransactionDefinition(TransactionDefinition.PROPAGATION_REQUIRES_NEW));
					for (LimsAsynctaskBean serverTask : updateRunnings) {
						serverTask.setServerId(currentServerId);
						serverTask.setExecStatus(AsyncTaskStatusEnum.executing.status);
						serverTask.setExecStartDate(LocalDateTime.now());
						serverTask.setExecCount(serverTask.getExecCount()+1);
						serverTask.setQueueProgress(null);
						serverTask.setErrMsg(null);
					}
					try {
						asynctaskService.getDao().fastUpdate(updateRunnings,"ServerId","ExecStatus","ExecStartDate","ExecCount","QueueProgress","ErrMsg");
						transactionManager.commit(transactionExecuting);
					} catch (Exception e) {
						logger.error("transactionManager error transactionExecuting:{}",e);
						transactionManager.rollback(transactionExecuting);
					}

					for (LimsAsynctaskBean updateRunning : updateRunnings) {
						asyncTaskExecutorHelper.submit(executorName,new AsyncTaskRunnable(updateRunning));
					}
				}
			}
		}
	}
}
